package com.zhb.cloud.core.distribute.zookeeper.lock;

import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListSet;

import org.apache.commons.io.Charsets;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.zhb.cloud.core.distribute.lock.DistributedLockUtils;

/**
 * @ClassName: ZooKeeperClientFactory
 * @author: zhb
 * @date: 2017年8月4日 上午11:05:30
 * @Description: zookeeper 连接工厂
 *
 */
public class ZooKeeperClientFactory {

	private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperClientFactory.class);

	/**
	 * session timeout in milliseconds
	 */
	public static final int ZOOKEEPER_SESSION_TIMEOUT = 10 * 1000;

	/**
	 * connection timeout in milliseconds
	 */
	public static final int ZOOKEEPER_CONNECTION_TIMEOUT = 10 * 1000;

	/**
	 * max number of times to retry
	 */
	public static final int ZOOKEEPER_RETRY_TIMES = 3;

	/**
	 * initial amount of time to wait between retries in milliseconds
	 */
	public static final int ZOOKEEPER_RETRY_INTERVAL = 1000;

	/**
	 * ZooKeeper clients cache
	 */
	private static final ConcurrentHashMap<String, CuratorFramework> sZKClientCache = new ConcurrentHashMap<String, CuratorFramework>();

	static {
		// shut down all zookeeper clients when exits
		Runtime.getRuntime().addShutdownHook(new Thread() {
			@Override
			public void run() {
				// <Configuration shutdownHook="disable"> in log4j2.xml to
				// support logger here
				closeZooKeeperClients();
			}
		});
	}

	/**
	 * ZooKeeper creation method
	 * 
	 * @param connectString
	 * @param namespace
	 * @return
	 */
	public static synchronized CuratorFramework createZooKeeperClient(String connectString, String namespace) {
		if (StringUtils.isEmpty(connectString)) {
			throw new NullPointerException("empty zookeeper connect string");
		}

		// normalize namespace
		if (!StringUtils.isBlank(namespace)) {
			namespace = namespace.replaceAll("^/+", "");
		}

		String keyClient = key(connectString, namespace);

		// return active client
		CuratorFramework client = sZKClientCache.get(keyClient);
		if (client == null) {
			// retry policy
			RetryPolicy retryPolicy = new ExponentialBackoffRetry(ZOOKEEPER_RETRY_INTERVAL, ZOOKEEPER_RETRY_TIMES);

			// default data: host address
			String host = DistributedLockUtils.getHostAddress();
			byte[] defaultData = null;
			if (host != null) {
				defaultData = host.getBytes(Charsets.toCharset("UTF-8"));
			}

			// build zk client
			CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder().connectString(connectString)
					.sessionTimeoutMs(ZOOKEEPER_SESSION_TIMEOUT).connectionTimeoutMs(ZOOKEEPER_CONNECTION_TIMEOUT)
					.canBeReadOnly(true).retryPolicy(retryPolicy).defaultData(defaultData);
			if (!StringUtils.isBlank(namespace)) {
				builder.namespace(namespace);
			}
			client = builder.build();
			LOG.info("Initializing zookeeper: connectString={} namespace={} session_timeout={} connection_timeout={}",
					connectString, namespace, ZOOKEEPER_SESSION_TIMEOUT, ZOOKEEPER_CONNECTION_TIMEOUT);

			sZKClientCache.put(keyClient, client);
		}

		// start zk client
		if (client.getState() != CuratorFrameworkState.STARTED) {
			client.start();
		}

		LOG.info("zookeeper client initialized succeed.");
		return client;
	}

	/**
	 * close zookeeper client
	 */
	public static synchronized void closeZooKeeperClient(CuratorFramework client) {
		client.close();

		// clear from cache
		for (Entry<String, CuratorFramework> entry : sZKClientCache.entrySet()) {
			sZKClientCache.remove(entry.getKey(), client);
		}
	}

	/**
	 * Close zookeeper client
	 */
	private static synchronized void closeZooKeeperClients() {
		for (Entry<String, CuratorFramework> entry : sZKClientCache.entrySet()) {
			entry.getValue().close();
			sZKClientCache.remove(entry.getKey());
		}
		LOG.info("zookeeper client destroy succeed.");
	}

	/**
	 * ZooKeeper client key
	 * 
	 * @param connectString
	 * @param namespace
	 * @return
	 */
	private static final String key(String connectString, String namespace) {
		if (StringUtils.isBlank(namespace))
			return connectString;
		return connectString + "_" + namespace;
	}

	/**
	 * GC path
	 */
	private static final ConcurrentSkipListSet<String> gcPaths = new ConcurrentSkipListSet<String>();

	private static final boolean ENABLE_GC = true;

	public static void gc(String path) {
		if (ENABLE_GC && !StringUtils.isEmpty(path)) {
			gcPaths.add(path);
		}
	}

}
